package cn._51doit.akka.rpc.我的练习

import akka.actor.{Actor, ActorSystem, Props}
import com.typesafe.config.ConfigFactory

import scala.collection.mutable
import scala.concurrent.duration._


/**
 *   ┏┓　　　┏┓
 * ┏┛┻━━━┛┻┓
 * ┃　　　　　　　┃ 　
 * ┃　　　━　　　┃
 * ┃　┳┛　┗┳　┃
 * ┃　　　　　　　┃
 * ┃　　　┻　　　┃
 * ┃　　　　　　　┃
 * ┗━┓　　　┏━┛
 *    　　┃　　　┃神兽保佑
 *    　　┃　　　┃代码无BUG！
 *    　　┃　　　┗━━━┓
 *    　　┃　　　　　　　┣┓
 *    　　┃　　　　　　　┏┛
 *    　　┗┓┓┏━┳┓┏┛
 *    　　　┃┫┫　┃┫┫
 *    　　　┗┻┛　┗┻┛
 *
 * @create: 2021-10-09 21:54
 * @author: 今晚打脑斧
 * @program: Master
 * @Description:
 *
 **/
class Master extends Actor {
  //创建一个Hashmap来存储worker(键值对方式存储)
  val id2Worker = new mutable.HashMap[String,WorkerInfo]()

  override def preStart(): Unit = {
    //启动一个定时器 定期检测超时的worker
    import context.dispatcher
    context.system.scheduler.schedule(0 second,6 second, self ,CheckTimeOutWorker)
  }
  //接收消息的方法
  override def receive:Receive = {
    case RegisterWorker(workerId,memory,cores)=>{
      //将worker的信息保存起来
      val workerInfo = new WorkerInfo(workerId, memory, cores)
      //将workerInfo保存到Map中
      id2Worker(workerId)=workerInfo
      //向Worker发送一个注册成功的消息
      sender() ! RegisteredWorker
    }
    case HeartBeat(workerId)=>{
      if (id2Worker.contains(workerId)){
        val workerInfo: WorkerInfo = id2Worker(workerId)
        workerInfo.lastHeartBeatTime=System.currentTimeMillis()
      }
    }
    case CheckTimeOutWorker=>{
      //检测超时的Worker
      val deadWorkers: Iterable[WorkerInfo] = id2Worker.values.filter(w => System.currentTimeMillis() - w.lastHeartBeatTime > 15000)
      deadWorkers.foreach(w=>id2Worker-=w.wokerId)
      println(s"current alive worker size is : ${id2Worker.size}")
    }

  }
}
object Master{
  val MASTER_ACTOR_SYSTEM_NAME = "MasterActorSystem"
  val MASTER_ACTOR_NAME = "MasterActor"
  def main(args: Array[String]): Unit = {
    val host = args(0)
    val port = args(1).toInt
    val configStr =
      s"""
        |akka.actor.provider = "akka.remote.RemoteActorRefProvider"
        |akka.remote.netty.tcp.hostname = $host
        |akka.remote.netty.tcp.port = $port
        |""".stripMargin
    val config = ConfigFactory.parseString(configStr)
    //创建ActorSystem(单例的)
    val masterActorSystem = ActorSystem(MASTER_ACTOR_SYSTEM_NAME,config)
    //创建Actor 通过actorof创建对应到的Actor
    masterActorSystem.actorOf(Props[Master],MASTER_ACTOR_NAME)
  }
}
